package com.rainsoft.center.isec.frame.start;

import com.rainsoft.center.isec.common.entity.Constants;
import com.rainsoft.center.isec.frame.entity.AppConfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;

/**
 * @Name com.rainsoft.center.isec.frame.start.PrepareSpark
 * @Description
 * @Author Elwyn
 * @Version 2017/12/8
 * @Copyright 上海云辰信息科技有限公司
 **/
public class PrepareSpark implements Serializable{
    private SparkConf sparkConf;
    private JavaSparkContext javaSparkContext;
    private JavaStreamingContext javaStreamingContext;
    private Map<String, Object> kafkaParams;
    private AppConfig appConfig;

    public PrepareSpark(AppConfig appConfig) {
        this.appConfig = appConfig;
        SparkConf conf = new SparkConf().setAppName(appConfig.getAppName());
        if (Constants.LOCAL) {
            conf.setMaster("local[2]");
        }
        this.sparkConf = conf;
        this.javaSparkContext = new JavaSparkContext(sparkConf);
        this.javaStreamingContext = new JavaStreamingContext(javaSparkContext, Durations.seconds(appConfig.getDurations()));
        this.kafkaParams = setKafkaConf();
    }

    private Map<String, Object> setKafkaConf() {
        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Constants.KAFKA_BROKERS);
        kafkaParams.put(ConsumerConfig.GROUP_ID_CONFIG, "group5");
        kafkaParams.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "30000");
        kafkaParams.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        kafkaParams.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        kafkaParams.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        return kafkaParams;
    }

    public Map<String, Object> getKafkaParams() {
        return kafkaParams;
    }

    public SparkConf getSparkConf() {
        return sparkConf;
    }


    public JavaSparkContext getJavaSparkContext() {
        return javaSparkContext;
    }


    public JavaStreamingContext getJavaStreamingContext() {
        return javaStreamingContext;
    }

    public AppConfig getAppConfig() {
        return appConfig;
    }
}
